"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 故障&可靠性测试
Case Name   : 构造备机扩容失败场景，故障恢复后再次扩容成功
Create At   : 2022/03/29
Owner       : n@ningyali
Description :
    1、创建测试表，插入测试数据
    2、构造备机扩容失败场景，比如，备机扩容过程中网络断连、备机断电
    3、再次进行备机扩容
    4、查看数据库状态
    5、查看测试表数据
Expect      :
    1、执行成功
    2、第一次扩容失败
    3、第二次扩容成功
    4、数据库集群状态正常
    5、数据一致
History     :
    modified: 2022/10/11  modified by wx1115623,扩容失败报错信息不固定，修改断言信息
"""
import os
import re
import time
import unittest
from datetime import datetime

from paramiko.ssh_exception import NoValidConnectionsError, SSHException
from yat.test import Node
from yat.test import macro

from testcase.utils.ComThread import ComThread
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH, RestartDbCluster
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger


class GsExpansion18(unittest.TestCase):
    nodes_tuple = ('PrimaryDbUser', 'Standby1DbUser', 'Standby2DbUser')
    log = Logger()

    @classmethod
    @RestartDbCluster(*nodes_tuple)
    def setUpClass(cls):
        text = f'-----{os.path.basename(__file__)} start-----'
        cls.log.info(text)

    def setUp(self):
        self.sh = {'pri_root': CommonSH('PrimaryRoot'),
                   'pri_user': CommonSH(self.nodes_tuple[0]),
                   'sta1_user': CommonSH(self.nodes_tuple[1]),
                   'sta1_root': CommonSH('Standby1Root'),
                   'sta2_user': CommonSH(self.nodes_tuple[2])}
        self.node_num = self.sh['pri_user'].get_node_num()
        self.cons = Constant()
        self.com = Common()
        self.network_card = ''
        self.t_name = 't_expansion_18'
        self.cfe_time = 900
        text = f'-----{os.path.basename(__file__)} run setup-----'
        self.log.info(text)

        text = '----预置步骤: 一主两备减容一备，后续用例才能扩容 expect: 成功----'
        self.log.info(text)
        expect_msg = 'Success to drop the target nodes'
        shell_cmd = f'''source {macro.DB_ENV_PATH};
                    expect <<EOF
                    set timeout 300
                    spawn gs_dropnode -U {self.sh['pri_user'].node.ssh_user} \
                    -G {self.sh['pri_user'].node.ssh_user} \
                    -h {self.sh.get("sta1_user").node.ssh_host}
                    expect "*(yes/no)?"
                    send "yes\\n"
                    expect eof''' + "\nEOF\n"
        self.log.info(shell_cmd)
        res = self.com.get_sh_result(self.sh['pri_user'].node, shell_cmd)
        self.assertIn(expect_msg, res, f'执行失败: {text}')

        text = '----预置步骤: 数据库主机解压安装包，以免找不到扩容脚本 expect: 成功----'
        self.log.info(text)
        cmd = f'cd {os.path.dirname(macro.DB_SCRIPT_PATH)} && ' \
            f'tar -xf openGauss-Package-bak*.tar.gz && ' \
            f'ls {macro.DB_SCRIPT_PATH}|grep gs_sshexkey'
        self.log.info(cmd)
        res = self.com.get_sh_result(self.sh.get("pri_root").node, cmd)
        self.assertEqual(res, 'gs_sshexkey', f'执行失败: {text}')

        text = '----预置步骤: 建立root互信 expect: 成功----'
        self.log.info(text)
        ssh_cmd = f'cd {macro.DB_SCRIPT_PATH};' \
            f'echo "{self.sh.get("pri_root").node.ssh_host}\n' \
            f'{self.sh.get("sta1_user").node.ssh_host}\n' \
            f'{self.sh.get("sta2_user").node.ssh_host}" > hosts\n'
        ssh_cmd += f'''expect <<EOF
                                    set timeout 300
                                    spawn ./gs_sshexkey -f hosts
                                    expect "*assword:"
                                    send "{self.sh.get(
            "pri_root").node.ssh_password}\\n"
                                    expect eof''' + "\nEOF\n"
        ssh_cmd += "source ~/.bashrc"
        self.log.info(ssh_cmd)
        res = self.com.get_sh_result(self.sh.get("pri_root").node, ssh_cmd)
        self.assertIn('Successfully created SSH trust', res, f'执行失败: {text}')

        text = '----step1:  创建测试表，插入测试数据 expect: 成功----'
        self.log.info(text)
        sql_cmd = f'drop table if exists {self.t_name};' \
            f'create table {self.t_name}(col1 int, col2 text);' \
            f'insert into {self.t_name} values ' \
            f'(generate_series(1,10000), generate_series(1,10000)||\'test\');'
        res = self.sh['pri_user'].executDbSql(sql_cmd)
        self.log.info(res)
        self.assertIn(self.cons.CREATE_TABLE_SUCCESS, res, f'执行失败: {text}')
        self.assertIn('INSERT 0 10000', res, f'执行失败: {text}')

    def test_1(self):
        text = '----step2: 构造备机扩容失败场景(备机扩容过程中网络断连) expect: 成功----'
        self.log.info(text)
        text = '----step2.1: 启线程执行备机扩容 expect: 执行失败----'
        self.log.info(text)
        expansion_cmd = f'cd {macro.DB_SCRIPT_PATH};' \
            f'source {macro.DB_ENV_PATH}\n' \
            f'./gs_expansion -U {self.sh.get("pri_user").node.ssh_user} ' \
            f'-G {self.sh.get("pri_user").node.ssh_user} ' \
            f'-X {macro.DB_XML_PATH} ' \
            f'-h {self.sh.get("sta1_user").node.ssh_host} -L'
        self.log.info(expansion_cmd)
        expansion_thread = ComThread(self.com.get_sh_result, args=(
            self.sh.get("pri_root").node, expansion_cmd,))
        expansion_thread.setDaemon(True)
        expansion_thread.start()

        time.sleep(5)

        text = '----step2.2: 获取网卡 expect: 执行成功----'
        self.log.info(text)
        shell_cmd = f"ifconfig|" \
            f"grep {self.sh['sta1_root'].node.ssh_host} -a2|" \
            f"head -1|awk -F ':' '{{{{print $1}}}}'"
        self.log.info(shell_cmd)
        msg = self.sh['sta1_root'].node.sh(shell_cmd).result()
        self.log.info(msg)
        self.network_card = msg.strip()
        self.assertNotEqual(len(self.network_card), 0, f'执行失败: {text}')

        text = '----step2.3: 停止DN主网卡30分钟 expect: 执行成功----'
        self.log.info(text)
        start_time = datetime.now()
        self.log.info(start_time)

        inject_command = f"rNet_down (dev, time)values(" \
            f"{self.network_card}, {self.cfe_time})"
        msg = self.com.cfe_inject(self.sh['sta1_root'].node, inject_command)
        self.log.info(msg)
        self.assertIn(self.cons.cfe_inject_netdown_success_msg, msg)

        text = '----step2.4: 获取备机扩容执行结果 expect: 获取结果成功，备机扩容失败，有合理报错----'
        self.log.info(text)
        expansion_thread.join(self.cfe_time * 2)
        res = expansion_thread.get_result()
        self.log.info(res)
        expect = 'Expansion results:.*:.*Success.*Expansion Finish.'
        regex_res = re.search(expect, res, re.S)
        self.assertIsNone(regex_res, f'执行失败: {text}')

        text = '----step3: 等待故障清除，再次进行备机扩容 expect: 第二次扩容成功----'
        self.log.info(text)
        end_time = datetime.now()
        self.log.info(end_time)
        need_wait_time = self.cfe_time - (end_time - start_time).seconds
        if need_wait_time > 0:
            time.sleep(need_wait_time)

        expansion_cmd = f'cd {macro.DB_SCRIPT_PATH};' \
            f'source {macro.DB_ENV_PATH}\n' \
            f'./gs_expansion -U {self.sh.get("pri_user").node.ssh_user} ' \
            f'-G {self.sh.get("pri_user").node.ssh_user} ' \
            f'-X {macro.DB_XML_PATH} ' \
            f'-h {self.sh.get("sta1_user").node.ssh_host} -L'
        self.log.info(expansion_cmd)
        res = self.com.get_sh_result(self.sh.get("pri_root").node,
                                     expansion_cmd)
        expect = 'Expansion results:.*:.*Success.*Expansion Finish.'
        regex_res = re.search(expect, res, re.S)
        self.assertIsNotNone(regex_res, f'执行失败: {text}')

        text = '----step4: 查看数据库状态 expect: 数据库集群状态正常----'
        self.log.info(text)
        res = self.sh['pri_user'].getDbClusterStatus('detail')
        self.log.info(res)
        self.assertEqual(res.count('Normal'), self.node_num + 1,
                         f'执行失败: {text}')

        text = '----step5: 查看测试表数据 expect: 数据一致----'
        self.log.info(text)
        nodes = (self.sh['pri_user'].node,
                 self.sh['sta1_user'].node,
                 self.sh['sta2_user'].node)
        flag = self.com.check_data_sample_by_all(r'\d', *nodes)
        self.assertTrue(flag)

        table_dict = self.com.format_sql_result(
            self.sh['pri_user'].executDbSql(r'\d'))
        table_name = table_dict.get('Name')
        for name in table_name:
            select_sql = f'select count(*) from {name};'
            flag = self.com.check_data_sample_by_all(select_sql,
                                                     *nodes)
            self.assertTrue(flag)

    def test_2(self):
        text = '----step2: 构造备机扩容失败场景(备机扩容过程中备机断电) expect: 成功----'
        self.log.info(text)
        text = '----step2.1: 启线程执行备机扩容 expect: 执行失败----'
        self.log.info(text)
        expansion_cmd = f'cd {macro.DB_SCRIPT_PATH};' \
            f'source {macro.DB_ENV_PATH}\n' \
            f'./gs_expansion -U {self.sh.get("pri_user").node.ssh_user} ' \
            f'-G {self.sh.get("pri_user").node.ssh_user} ' \
            f'-X {macro.DB_XML_PATH} ' \
            f'-h {self.sh.get("sta1_user").node.ssh_host} -L'
        self.log.info(expansion_cmd)
        expansion_thread = ComThread(self.com.get_sh_result, args=(
            self.sh.get("pri_root").node, expansion_cmd,))
        expansion_thread.setDaemon(True)
        expansion_thread.start()

        time.sleep(5)

        text = '----step2.2: 备机扩容过程中备机断电(因单次reboot时间较短，此处做多次reboot操作) ' \
               'expect: 执行成功----'
        self.log.info(text)
        for i in range(self.cfe_time // 30):
            sta1_root_node = Node('Standby1Root')
            try:
                with self.assertRaises(
                        (SSHException, NoValidConnectionsError, EOFError)):
                    self.com.get_sh_result(sta1_root_node, 'echo T')
            except Exception as e:
                self.log.info(str(e))
                self.log.info('备机已启动，故障已清除，再次注入故障')
                self.com.get_sh_result(sta1_root_node, 'reboot')
            else:
                time.sleep(30)

        text = '----step2.3: 获取备机扩容执行结果 expect: 获取结果成功，备机扩容失败，有合理报错----'
        self.log.info(text)
        expansion_thread.join(self.cfe_time * 2)
        res = expansion_thread.get_result()
        self.log.info(res)
        expect = 'Expansion results:.*:.*Success.*Expansion Finish.'
        regex_res = re.search(expect, res, re.S)
        self.assertIsNone(regex_res, f'执行失败: {text}')

        text = '----step3: 等待故障清除，再次进行备机扩容 expect: 第二次扩容成功----'
        self.log.info(text)
        for i in range(self.cfe_time // 30):
            sta1_root_node = Node('Standby1Root')
            try:
                with self.assertRaises(
                        (SSHException, NoValidConnectionsError, EOFError)):
                    self.com.get_sh_result(sta1_root_node, 'echo T')
            except Exception as e:
                self.log.info(str(e))
                self.log.info('备机已启动，故障已清除')
                break
            else:
                time.sleep(30)

        expansion_cmd = f'cd {macro.DB_SCRIPT_PATH};' \
            f'source {macro.DB_ENV_PATH}\n' \
            f'./gs_expansion -U {self.sh.get("pri_user").node.ssh_user} ' \
            f'-G {self.sh.get("pri_user").node.ssh_user} ' \
            f'-X {macro.DB_XML_PATH} ' \
            f'-h {self.sh.get("sta1_user").node.ssh_host} -L'
        self.log.info(expansion_cmd)
        res = self.com.get_sh_result(self.sh.get("pri_root").node,
                                     expansion_cmd)
        expect = 'Expansion results:.*:.*Success.*Expansion Finish.'
        regex_res = re.search(expect, res, re.S)
        self.assertIsNotNone(regex_res, f'执行失败: {text}')

        text = '----step4: 查看数据库状态 expect: 数据库集群状态正常----'
        self.log.info(text)
        res = self.sh['pri_user'].getDbClusterStatus('detail')
        self.log.info(res)
        self.assertEqual(res.count('Normal'), self.node_num + 1,
                         f'执行失败: {text}')

        text = '----step5: 查看测试表数据 expect: 数据一致----'
        self.log.info(text)
        nodes = (self.sh['pri_user'].node,
                 self.sh['sta1_user'].node,
                 self.sh['sta2_user'].node)
        flag = self.com.check_data_sample_by_all(r'\d', *nodes)
        self.assertTrue(flag)

        table_dict = self.com.format_sql_result(
            self.sh['pri_user'].executDbSql(r'\d'))
        table_name = table_dict.get('Name')
        for name in table_name:
            select_sql = f'select count(*) from {name};'
            flag = self.com.check_data_sample_by_all(select_sql,
                                                     *nodes)
            self.assertTrue(flag)

    def tearDown(self):
        text = f'-----{os.path.basename(__file__)} run teardown-----'
        self.log.info(text)

        text1 = f'----删除表----'
        self.log.info(text1)
        sql_cmd = f'drop table if exists {self.t_name};'
        res1 = self.sh['pri_user'].executDbSql(sql_cmd)
        self.log.info(res1)

        text2 = f'----检查集群状态是否正常----'
        self.log.info(text2)
        res2 = self.sh['pri_user'].getDbClusterStatus('detail')
        self.log.info(res2)

        self.assertIn(self.cons.DROP_TABLE_SUCCESS, res1, f'执行失败: {text1}')
        self.assertEqual(res2.count('Normal'), self.node_num + 1,
                         f'执行失败: {text2}')

    @classmethod
    def tearDownClass(cls):
        text = f'-----{os.path.basename(__file__)} end-----'
        cls.log.info(text)
